/*
    消息队列相关类的定义：消息队列类、消息队列管理类
    1. 消息队列类：对消息队列的抽象描述
        属性：名称、持久化标志、独占标志、自动删除标志、其他参数
        方法：创建消息队列、设置其他参数、获取其他参数
    2. 消息队列管理类：对消息队列数据进行管理以及持久化
        属性：数据库文件(消息队列数据持久化的目标数据库文件)、消息队列数据映射表(<消息队列名, 消息队列对象>)
        方法：声明消息队列、删除消息队列、获取消息队列、获取所有消息队列、判断消息队列是否存在、销毁所有消息队列数据
*/
#pragma once
#include "../common/helper.hpp"
#include "../common/logger.hpp"
#include <google/protobuf/map.h>
#include <string>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <sstream>
#include <vector>


namespace mq
{
    // 消息队列类
    struct MsgQueue
    {
        MsgQueue(){}
        MsgQueue(const std::string& name, bool durable, bool exclusive, bool auto_delete, const google::protobuf::Map<std::string, std::string>& args):
            _name(name), _durable(durable), _exclusive(exclusive), _auto_delete(auto_delete), _args(args)
        {
        }

        // 获取其他参数
        std::string getArgs()
        {
            std::string ret;
            for(auto& e : _args)
            {
                ret += e.first + "=" + e.second + "&";
            }
            if(ret.size())
                ret.pop_back();
            return ret;
        }

        // 设置其他参数
        void setArgs(const std::string& args_str)
        {
            std::vector<std::string> kvs;
            StringHelper::split(args_str, "&", kvs);
            for(auto& kv : kvs)
            {
                int pos = kv.find("=");
                std::string key = kv.substr(0, pos);
                std::string value = kv.substr(pos+1);
                _args[key] = value;
            }
        }

        std::string _name; // 名称
        bool _durable; // 持久化标志
        bool _exclusive; // 独占标志
        bool _auto_delete; // 自动删除标志
        google::protobuf::Map<std::string, std::string> _args; // 其他参数
        using MsgQueuePtr = std::shared_ptr<MsgQueue>;
    };

    // 消息队列管理类
    class MsgQueueManager
    {
    public:
        using MsgQueueManagerPtr = std::shared_ptr<MsgQueueManager>;
        using MsgQueueMap = std::unordered_map<std::string, MsgQueue::MsgQueuePtr>;

    public:
        MsgQueueManager(const std::string& db_file): _sqlite3_helper(db_file)
        {
            std::string parent_dir = FileHelper::parentDir(db_file);
            if(!FileHelper::createDir(parent_dir))
                ERR_LOG("create dir %s failed\n", parent_dir.c_str());
            if(!_sqlite3_helper.open())
                ERR_LOG("open message queue db %s failed\n", db_file.c_str());
            createTable();
            _msg_queues = recoveryAllMsgQueues();
        }

        // 声明消息队列
        bool declareMsgQueue(const std::string& name, bool durable, bool exclusive, bool auto_delete, const google::protobuf::Map<std::string, std::string>& args)
        {
            std::lock_guard<std::mutex> lock(_mutex);
            if(_msg_queues.count(name))
                return true;
            MsgQueue::MsgQueuePtr msg_queue_ptr = std::make_shared<MsgQueue>(name, durable, exclusive, auto_delete, args);
            if(durable && !insertMsgQueue(msg_queue_ptr))
            {
                ERR_LOG("declare message queue %s failed\n", name.c_str());
                return false;
            }
            _msg_queues[name] = msg_queue_ptr;
            return true;
        }

        // 删除消息队列
        void removeMsgQueue(const std::string& name)
        {
            std::lock_guard<std::mutex> lock(_mutex);
            auto it = _msg_queues.find(name);
            if(it == _msg_queues.end())
                return;
            if(it->second->_durable)
                deleteMsgQueue(name);
            _msg_queues.erase(it);
        }

        // 获取消息队列
        MsgQueue::MsgQueuePtr getMsgQueue(const std::string& name)
        {
            std::lock_guard<std::mutex> lock(_mutex);
            auto it = _msg_queues.find(name);
            if(it == _msg_queues.end())
                return nullptr;
            return it->second;
        }

        // 获取所有消息队列
        MsgQueueMap getAllMsgQueues()
        {
            std::lock_guard<std::mutex> lock(_mutex);
            return _msg_queues;
        }

        // 判断消息队列是否存在
        bool hasMsgQueue(const std::string& name)
        {
            std::lock_guard<std::mutex> lock(_mutex);
            return _msg_queues.count(name);
        }

        // 获取消息队列数量
        size_t size()
        {
            std::lock_guard<std::mutex> lock(_mutex);
            return _msg_queues.size();
        }

        // 清空消息队列
        void clear()
        {
            std::lock_guard<std::mutex> lock(_mutex);
            deleteTable();
            _msg_queues.clear();
        }

    private:
        /*
            消息队列持久化相关的操作
        */
       // 创建表
        void createTable()
        {
            #define MSG_QUEUE_CREATE_TABLE_SQL "create table if not exists queue(name varchar(32) primary key, durable int, exclusive int, auto_delete int, args varchar(1024))"
            if(!_sqlite3_helper.execute(MSG_QUEUE_CREATE_TABLE_SQL, nullptr, nullptr))
            {
                ERR_LOG("create table queue failed\n");
                abort();
            }
        }

        // 删除表
        void deleteTable()
        {
            #define MSG_QUEUE_DELETE_TABLE_SQL "drop table if exists queue"
            if(!_sqlite3_helper.execute(MSG_QUEUE_DELETE_TABLE_SQL, nullptr, nullptr))
            {
                ERR_LOG("drop table queue failed\n");
                abort();
            }
        }

        // 插入数据
        bool insertMsgQueue(const MsgQueue::MsgQueuePtr& msg_queue_ptr)
        {
            if(msg_queue_ptr == nullptr)
            {
                DBG_LOG("insert queue failed, msg_queue_ptr is nullptr\n");
                return false;
            }
            std::stringstream sql;
            sql << "insert into queue(name, durable, exclusive, auto_delete, args) values(";
            sql << "'" << msg_queue_ptr->_name << "'," << msg_queue_ptr->_durable << ",";
            sql << msg_queue_ptr->_exclusive << "," << msg_queue_ptr->_auto_delete << ",";
            sql << "'" << msg_queue_ptr->getArgs() << "');";
            if (!_sqlite3_helper.execute(sql.str(), nullptr, nullptr))
            {
                ERR_LOG("insert queue failed\n");
                return false;
            }
            return true;
        }

        // 删除数据
        void deleteMsgQueue(const std::string& msg_queue_name)
        {
            std::stringstream sql;
            sql << "delete from queue where name = '" << msg_queue_name << "';";
            if (!_sqlite3_helper.execute(sql.str(), nullptr, nullptr))
                ERR_LOG("delete %s queue failed\n", msg_queue_name.c_str());
        }

        // 恢复数据
        MsgQueueMap recoveryAllMsgQueues()
        {
            MsgQueueMap msg_queue_map;
            if(!_sqlite3_helper.execute("select * from queue;", selectCallback, &msg_queue_map))
                ERR_LOG("get message queue all failed\n");
            return msg_queue_map;
        }

        // 回调函数
        static int selectCallback(void* arg, int col, char** values, char** files)
        {
            MsgQueueMap* msg_queue_map_ptr = reinterpret_cast<MsgQueueMap*>(arg);
            MsgQueue::MsgQueuePtr msg_queue_ptr = std::make_shared<MsgQueue>();
            msg_queue_ptr->_name = values[0];
            msg_queue_ptr->_durable = std::stoi(values[1]);
            msg_queue_ptr->_exclusive = std::stoi(values[2]);
            msg_queue_ptr->_auto_delete = std::stoi(values[3]);
            msg_queue_ptr->setArgs(values[4]);
            (*msg_queue_map_ptr)[msg_queue_ptr->_name] = msg_queue_ptr;
            return 0;
        }

    private:
        MsgQueueMap _msg_queues;
        std::mutex _mutex;
        Sqlite3Helper _sqlite3_helper;
    };
}